package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.cps.processors.vector.SchemaUtility;
import com.ibm.cps.sql.expression.BooleanExpression;
import com.ibm.cps.sql.expression.FieldEvaluator;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.rules.RuleMode;
import com.ibm.util.DataTypes;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.NAIVE_RULE,Type=Info.ProcessorType.Other)
public class NaiveRuleEngineProcessor extends SchemaInheritiveProcessor {
	public static final String RULE_NAME = "RULE_NAME";
	
	@ValueParam @IgnoredInFunctionCall
	private String ruleName;
	@ValueParam @IgnoredInFunctionCall
	private String expressionNodeClassName;
	@ValueParam @IgnoredInFunctionCall
	private String expressionJson;
	@ValueParam @IgnoredInFunctionCall
	private String ruleModeCode;
	
	private BooleanExpression rule;
	private Set<String> usedFields;
	private RuleMode ruleMode;
	private Map<Comparable, Boolean> ruleState = new HashMap<>();
	
	public NaiveRuleEngineProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	}
	
	

	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		List<NewField> newFields = new ArrayList<>();
		newFields.add(new NewField(RULE_NAME, "string"));
		newFields.add(new NewField("hitten", "string"));
		return newFields;
	}

	@Override
	public void dispose() {
	}

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		
		FieldEvaluator evaluator = SchemaUtility.createEvaluator(parentSchema, message, usedFields);
		boolean currentState = rule.evaluate(evaluator);
		Comparable tsKey = message.getTsKey();
		if(ruleMode.shouldEmit(ruleState.get(tsKey), currentState)){
			outputBuilder.append(RULE_NAME, ruleName);
			outputBuilder.append("hitten", currentState);
		}
		ruleState.put(tsKey, currentState);
		
		messages.add(outputBuilder.toParsedDataMessage());
		return messages;
	}

	@Override
	public void initialize() throws CPSException {
		this.rule = JSONUtility.fromJsonToObjectWithClassName(expressionJson, expressionNodeClassName);
		this.usedFields = rule.collectVariableNames();
		this.ruleMode = RuleMode.valueOf(ruleModeCode);
	}

}
